[log] FWaaS L3 Logging driver based iptables

This patch implements logging driver in L3 for firewall group
base discussed on the patch [1]

[1] https://review.openstack.org/#/c/509725/

Co-Authored-By: Nguyen Phuong An <AnNP@vn.fujitsu.com>
Co-Authored-By: Kim Bao Long <longkb@vn.fujitsu.com>
Partial-Bug: #1720727
Change-Id: I1194a622c546068991f44559e3f9e343430fd6f9

doc/source/conf.py

@@ -61,7 +61,8 @@ todo_include_todos = True
 # sphinxcontrib.apidoc options
 apidoc_module_dir = '../../neutron_fwaas'
 apidoc_output_dir = 'contributor/api'
-# TODO(hoangcx): remove 'services/logapi/*' after next neutron release
+# TODO(hoangcx): remove 'services/logapi/*' and
+# 'services/firewall/fwaas_plugin_v2.py' after the next neutron release
 # (current release is Rocky-3)
 
 # NOTE(longkb): Due to libnetfilter_log library is not installed in sphinx-docs
@@ -70,6 +71,7 @@ apidoc_output_dir = 'contributor/api'
 apidoc_excluded_paths = [
     'db/migration/alembic_migrations/*',
     'privileged/netfilter_log/*',
+    'services/firewall/fwaas_plugin_v2.py',
     'services/logapi/*',
     'setup.py',
     'tests/*',

neutron_fwaas/services/firewall/fwaas_plugin_v2.py

@@ -24,6 +24,7 @@ from neutron_lib.callbacks import registry
 from neutron_lib.callbacks import resources
 from neutron_lib import constants as nl_constants
 from neutron_lib.exceptions import firewall_v2 as f_exc
+from neutron_lib.plugins import constants as plugin_const
 from neutron_lib.plugins import directory
 from oslo_log import helpers as log_helpers
 from oslo_log import log as logging
@@ -32,6 +33,8 @@ from neutron_fwaas.common import exceptions
 from neutron_fwaas.common import fwaas_constants
 from neutron_fwaas.extensions.firewall_v2 import Firewallv2PluginBase
 from neutron_fwaas.services.firewall.service_drivers import driver_api
+from neutron_fwaas.services.logapi.agents.drivers.iptables \
+    import driver as logging_driver
 
 LOG = logging.getLogger(__name__)
 
@@ -70,6 +73,13 @@ class FirewallPluginV2(Firewallv2PluginBase):
             rpc_worker = service.RpcWorker([self], worker_process_count=0)
             self.add_worker(rpc_worker)
 
+        log_plugin = directory.get_plugin(plugin_const.LOG_API)
+        logging_driver.register()
+        # If log_plugin was loaded before firewall plugin
+        if log_plugin:
+            # Register logging driver with LoggingServiceDriverManager again
+            log_plugin.driver_manager.register_driver(logging_driver.DRIVER)
+
     def start_rpc_listeners(self):
         return self.driver.start_rpc_listener()
 

neutron_fwaas/services/logapi/agents/drivers/iptables/driver.py

@@ -0,0 +1,67 @@
+# Copyright (c) 2018 Fujitsu Limited.
+# 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.services.logapi.common import constants as log_const
+from neutron.services.logapi.drivers import base
+from neutron.services.logapi.drivers import manager
+from neutron_lib.callbacks import resources
+from oslo_log import log as logging
+from oslo_utils import importutils
+
+from neutron_fwaas.common import fwaas_constants
+from neutron_fwaas.services.logapi.common import fwg_callback
+from neutron_fwaas.services.logapi import constants as fw_const
+from neutron_fwaas.services.logapi.rpc import log_server as rpc_server
+
+LOG = logging.getLogger(__name__)
+
+DRIVER = None
+
+SUPPORTED_LOGGING_TYPES = [fw_const.FIREWALL_GROUP]
+
+
+class IptablesLoggingDriver(base.DriverBase):
+
+    @staticmethod
+    def create():
+        return IptablesLoggingDriver(
+            name='iptables',
+            vif_types=[],
+            vnic_types=[],
+            supported_logging_types=SUPPORTED_LOGGING_TYPES,
+            requires_rpc=True)
+
+
+def register():
+    """Register iptables-based logging driver for FWaaS."""
+
+    global DRIVER
+    if not DRIVER:
+        DRIVER = IptablesLoggingDriver.create()
+        # Register RPC methods
+        if DRIVER.requires_rpc:
+            rpc_methods = [
+                {resources.PORT: rpc_server.get_fwg_log_info_for_port},
+                {log_const.LOG_RESOURCE: rpc_server.
+                    get_fwg_log_info_for_log_resources}
+            ]
+            DRIVER.register_rpc_methods(fw_const.FIREWALL_GROUP, rpc_methods)
+
+    # Trigger fwg validator
+    importutils.import_module('neutron_fwaas.services.logapi.fwg_validate')
+    # Register resource callback handler
+    manager.register(
+        fwaas_constants.FIREWALL_GROUP, fwg_callback.FirewallGroupCallBack)
+    LOG.debug('FWaaS L3 Logging driver based iptables registered')

neutron_fwaas/services/logapi/agents/drivers/iptables/log.py

@@ -0,0 +1,490 @@
+# Copyright (c) 2018 Fujitsu Limited.
+# 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 collections import defaultdict
+import signal
+import uuid
+
+from neutron.agent.linux import utils
+from neutron.common import constants as n_const
+from neutron.services.logapi.agent import log_extension as log_ext
+from neutron.services.logapi.common import constants as log_const
+from neutron_lib import constants
+from oslo_config import cfg
+from oslo_log import handlers
+from oslo_log import log as logging
+
+from neutron_fwaas.privileged.netfilter_log import libnetfilter_log as libnflog
+
+LOG = logging.getLogger(__name__)
+
+UINT64_BITMASK = (1 << 64) - 1
+
+MAX_INTF_NAME_LEN = 14
+INTERNAL_DEV_PREFIX = 'qr-'
+SNAT_INT_DEV_PREFIX = 'sg-'
+ROUTER_2_FIP_DEV_PREFIX = 'rfp-'
+
+IPTABLES_DIRECTION_DEVICE = {
+    constants.INGRESS_DIRECTION: 'i',
+    constants.EGRESS_DIRECTION: 'o'
+}
+
+
+def setup_logging():
+
+    log_file = cfg.CONF.network_log.local_output_log_base
+    if log_file:
+        import logging
+        log_file_handler = logging.handlers.WatchedFileHandler(log_file)
+        log_file_handler.setLevel(logging.INFO)
+        log_file_handler.setFormatter(logging.Formatter(
+            fmt="%(asctime)s %(message)s", datefmt=cfg.CONF.log_date_format))
+        LOG.logger.addHandler(log_file_handler)
+    elif cfg.CONF.use_journal:
+        journal_handler = handlers.OSJournalHandler()
+        LOG.logger.addHandler(journal_handler)
+    else:
+        syslog_handler = handlers.OSSysLogHandler()
+        LOG.logger.addHandler(syslog_handler)
+
+
+class LogPrefix(object):
+    """LogPrefix could be used as prefix in NFLOG rules
+    Each of a couple (port_id, event) has its own LogPrefix object
+    """
+
+    def __init__(self, port_id, event, project_id):
+        self.id = self._generate_prefix_id()
+        self.port_id = port_id
+        self.action = event
+        # A list of log objects that referenced to this prefix
+        self.log_object_refs = set()
+        self.project_id = project_id
+
+    def __eq__(self, other):
+        return (self.id == other.id and
+                self.action == other.action and
+                self.port_id == other.port_id)
+
+    def __hash__(self):
+        return hash(self.id)
+
+    def _generate_prefix_id(self):
+        return uuid.uuid4().int & UINT64_BITMASK
+
+    def add_log_obj_ref(self, log_id):
+        self.log_object_refs.add(log_id)
+
+    def remove_log_obj_ref(self, log_id):
+        self.log_object_refs.discard(log_id)
+
+    @property
+    def is_empty(self):
+        return not self.log_object_refs
+
+
+class FWGPortLog(object):
+    """A firewall group port log per log_object"""
+
+    def __init__(self, port_id, log_info):
+        self.port_id = port_id
+        self.log_id = log_info['id']
+        self.project_id = log_info['project_id']
+        self.event = log_info['event']
+
+
+class IptablesLoggingDriver(log_ext.LoggingDriver):
+
+    SUPPORTED_LOGGING_TYPES = ['firewall_group']
+
+    def __init__(self, agent_api):
+        self.agent_api = agent_api
+        self.conf = cfg.CONF
+        self.rate_limit = self.conf.network_log.rate_limit
+        if self.rate_limit:
+            self.burst_limit = self.conf.network_log.burst_limit
+        self.ipt_mgr_list = defaultdict(dict)
+        # A list of fwg port logs that are being logged
+        self.fwg_port_logs = defaultdict(set)
+        # A list of prefixes that are being used in iptables
+        self.prefixes_table = {}
+        self.cleanup_table = defaultdict(set)
+        # Handle NFLOG processing
+        self.nflog_proc_map = {}
+
+    def initialize(self, resource_rpc, **kwargs):
+        self.resource_rpc = resource_rpc
+        setup_logging()
+        self.log_app = libnflog.NFLogApp()
+        self.log_app.register_packet_handler(self.log_packet)
+        self.log_app.start()
+
+    def log_packet(self, ev):
+        prefix = ev['prefix']
+        pkt = ev['msg']
+        prefix_entry = self._get_prefix_by_id(prefix)
+        if prefix_entry:
+            logs_id = [str(id) for id in prefix_entry.log_object_refs]
+            LOG.info("action=%s, project_id=%s, log_resource_ids=%s, port=%s, "
+                     "pkt=%s", prefix_entry.action,
+                     prefix_entry.project_id, logs_id,
+                     prefix_entry.port_id, pkt)
+        else:
+            LOG.warning("Unknown cookie packet_in pkt=%s", pkt)
+        return 0
+
+    def _get_prefix(self, port_id, action):
+        if port_id in self.prefixes_table:
+            for prefix in self.prefixes_table[port_id]:
+                if prefix.action == action:
+                    return prefix
+        return None
+
+    def _get_prefix_by_id(self, prefix_id):
+        for port, prefixes in self.prefixes_table.items():
+            for prefix in prefixes:
+                if str(prefix.id) == str(prefix_id):
+                    return prefix
+        return None
+
+    def _add_to_cleanup(self, port_id, prefix_id):
+        if port_id not in self.cleanup_table:
+            self.cleanup_table[port_id] = set()
+        self.cleanup_table[port_id].add(prefix_id)
+
+    def _add_to_prefixes_table(self, port_id, prefix):
+        if port_id not in self.prefixes_table:
+            self.prefixes_table[port_id] = []
+        self.prefixes_table[port_id].append(prefix)
+
+    def _cleanup_nflog_process(self, router_info):
+        LOG.debug("Delete router_info %s", router_info)
+        if router_info in self.nflog_proc_map:
+            pid = self.nflog_proc_map[router_info]
+            try:
+                # A process started by a root helper will be running as
+                # root and need to be killed via the same helper.
+                LOG.debug('Trying to kill NFLOG process %d', pid)
+                utils.kill_process(pid, signal.SIGKILL, run_as_root=True)
+                del self.nflog_proc_map[router_info]
+            except Exception:
+                LOG.exception(
+                    'An error occurred while killing process %d', pid)
+
+    def _cleanup_prefix_by_router(self, router_id):
+
+        ipt_mgr_per_port = set()
+        for port_id in self.ipt_mgr_list[router_id]:
+            ipt_mgr = self.ipt_mgr_list[router_id][port_id]
+            ipt_mgr_per_port.add(ipt_mgr)
+            # Cleanup prefix
+            if port_id in self.prefixes_table:
+                for prefix in self.prefixes_table[port_id]:
+                    self._add_to_cleanup(port_id, prefix.id)
+                del self.prefixes_table[port_id]
+        return ipt_mgr_per_port
+
+    def start_logging(self, context, **kwargs):
+        LOG.debug("Start logging: %s", str(kwargs))
+
+        for resource_type in self.SUPPORTED_LOGGING_TYPES:
+            router_info = kwargs.get('router_info')
+            if router_info:
+                # Handle router updated or L3 agent restart
+                router_id = router_info.router_id
+                internal_ports = router_info.internal_ports
+                self._create_firewall_group_log(context, resource_type,
+                                                ports=internal_ports,
+                                                router_id=router_id)
+
+                # Start libnetfilter_log after router starting up
+                pid = libnflog.run_nflog(router_info.ns_name)
+                LOG.debug("NFLOG process ID %s for router %s has started",
+                        pid, router_info.router_id)
+                self.nflog_proc_map[router_id] = pid
+            else:
+                # Handle the log request
+                self._create_firewall_group_log(context, resource_type,
+                                                **kwargs)
+
+    def stop_logging(self, context, **kwargs):
+        LOG.debug("Stop logging: %s", str(kwargs))
+
+        # Delete router
+        router_info = kwargs.get('router_info')
+        if router_info:
+            self._cleanup_nflog_process(router_info)
+
+        if kwargs.get('log_resources'):
+            # Handle incoming log request
+            self._delete_firewall_group_log(context, **kwargs)
+
+    def _create_firewall_group_log(self, context, resource_type, **kwargs):
+        ports = kwargs.get('ports')
+        log_resources = kwargs.get('log_resources')
+        applied_ipt_mgrs = set()
+        logs_info = []
+
+        port_ids = []
+        # Get log objects from database via RPC
+        if ports:
+            port_ids = [port['id'] for port in ports]
+            logs_info = self.resource_rpc. \
+                get_sg_log_info_for_port(context, resource_type,
+                                         port_id=port_ids)
+        elif log_resources:
+            logs_info = self.resource_rpc.\
+                get_sg_log_info_for_log_resources(context, resource_type,
+                                                  log_resources=log_resources)
+        # Handle logs_info
+        for log_info in logs_info:
+            log_id = log_info['id']
+            old_fwg_port_logs = self.fwg_port_logs.get(log_id, [])
+            new_ports_log = log_info.get('ports_log')
+            self.fwg_port_logs[log_id] = set()
+            for port in new_ports_log:
+                self._add_fwg_port_log(log_id, port, log_info)
+
+            for port in old_fwg_port_logs:
+                if port.port_id not in new_ports_log:
+                    # Remove port not bound by log_id
+                    self._cleanup_prefixes_table(port.port_id, log_id)
+
+            for fwg_port_log in self.fwg_port_logs[log_id]:
+                self._setup_chains(applied_ipt_mgrs, fwg_port_log)
+
+        router_id = kwargs.get("router_id")
+        if router_id:
+            if not port_ids:
+                ipt_mgrs = self._cleanup_prefix_by_router(router_id)
+                applied_ipt_mgrs.update(ipt_mgrs)
+
+            for port_id in port_ids:
+                try:
+                    ipt_mgr = self.ipt_mgr_list[router_id][port_id]
+                    applied_ipt_mgrs.add(ipt_mgr)
+                except KeyError:
+                    pass
+
+        # Clean up NFLOG rules
+        self._cleanup_nflog_rules(applied_ipt_mgrs)
+        # Apply NFLOG rules into iptables managers
+        for ipt_mgr in applied_ipt_mgrs:
+            LOG.debug('Apply NFLOG rules in namespace %s', ipt_mgr.namespace)
+            ipt_mgr.defer_apply_off()
+
+    def _cleanup_prefixes_table(self, port_id, log_id):
+
+        # Each a port has at most 2 prefix
+        for index in [1, 0]:
+            try:
+                prefix = self.prefixes_table[port_id][index]
+                prefix.remove_log_obj_ref(log_id)
+                if prefix.is_empty:
+                    self._add_to_cleanup(port_id, prefix.id)
+                    self.prefixes_table[port_id].remove(prefix)
+            except KeyError:
+                pass
+
+        if port_id in self.prefixes_table:
+            del self.prefixes_table[port_id]
+
+    def _cleanup_nflog_rules(self, applied_ipt_mgrs):
+        for port_id, prefix_ids in self.cleanup_table.items():
+            ipt_mgr = self._get_ipt_mgr_by_port(port_id)
+            for prefix_id in prefix_ids:
+                self._clear_rules_from_tag_v4v6(ipt_mgr, tag=prefix_id)
+            applied_ipt_mgrs.add(ipt_mgr)
+        self.cleanup_table.clear()
+
+    def _delete_firewall_group_log(self, context, **kwargs):
+        log_resources = kwargs.get('log_resources')
+        applied_ipt_mgrs = set()
+
+        for log_resource in log_resources:
+            log_id = log_resource.get('id')
+            fwg_port_logs = self.fwg_port_logs[log_id]
+            for port in fwg_port_logs:
+                self._cleanup_prefixes_table(port.port_id, log_id)
+            del self.fwg_port_logs[log_id]
+
+        # Clean NFLOG rules:
+        self._cleanup_nflog_rules(applied_ipt_mgrs)
+        # Apply NFLOG rules into iptables managers
+        for ipt_mgr in applied_ipt_mgrs:
+            ipt_mgr.defer_apply_off()
+
+    def _get_if_prefix(self, agent_mode, router):
+        """Get the if prefix from router"""
+        if not router.router.get('distributed'):
+            return INTERNAL_DEV_PREFIX
+
+        if agent_mode == 'dvr_snat':
+            return SNAT_INT_DEV_PREFIX
+
+        if router.rtr_fip_connect:
+            return ROUTER_2_FIP_DEV_PREFIX
+
+    def _get_intf_name(self, port_id):
+        agent_mode = self.conf.agent_mode
+        router = self.agent_api.get_router_hosting_port(port_id)
+        if_prefix = self._get_if_prefix(agent_mode, router)
+        return (if_prefix + port_id)[:n_const.LINUX_DEV_LEN]
+
+    def _get_ipt_mgr_by_port(self, port_id):
+
+        router = self.agent_api.get_router_hosting_port(port_id)
+        if router:
+            try:
+                ipt_mgr = self.ipt_mgr_list[router.router_id][port_id]
+                return ipt_mgr
+            except KeyError:
+                pass
+
+            ipt_mgr = router.iptables_manager
+            self.ipt_mgr_list[router.router_id][port_id] = ipt_mgr
+            return ipt_mgr
+
+        for router_id in self.ipt_mgr_list:
+            if port_id in self.ipt_mgr_list[router_id]:
+                return self.ipt_mgr_list[router_id][port_id]
+
+    def _setup_chains(self, applied_ipt_mgrs, fwg_port_log):
+        # Add NFLOG rules by log event
+        event = fwg_port_log.event
+        if event in [log_const.ACCEPT_EVENT, log_const.ALL_EVENT]:
+            self._add_nflog_rules_accepted(applied_ipt_mgrs, fwg_port_log)
+        if event in [log_const.DROP_EVENT, log_const.ALL_EVENT]:
+            self._add_log_rules_dropped(applied_ipt_mgrs, fwg_port_log)
+
+    def _add_nflog_rules_accepted(self, applied_ipt_mgrs, fwg_port_log):
+        """Add NFLOG rules to the accepted chain into iptables"""
+        action = log_const.ACCEPT_EVENT
+        port_id = fwg_port_log.port_id
+        prefix = self._get_prefix(port_id, action)
+        if not prefix:
+            # Generate a new prefix from port and action
+            project_id = fwg_port_log.project_id
+            prefix = LogPrefix(port_id, action, project_id)
+            self._add_to_prefixes_table(port_id, prefix)
+
+            # Get iptables manager from router port
+            ipt_mgr = self._get_ipt_mgr_by_port(port_id)
+            if ipt_mgr:
+                applied_ipt_mgrs.add(ipt_mgr)
+
+            device = self._get_intf_name(port_id)
+
+            # Add the NFLOG rules to the dropped chain into iptables
+            ipv4_rules, ipv6_rules = \
+                self._generate_nflog_rules_v4v6(device, prefix=prefix.id)
+            self._add_rules_to_chain_v4v6(ipt_mgr,
+                                          'accepted', ipv4_rules, ipv6_rules,
+                                          wrap=True, top=True, tag=prefix.id)
+
+        prefix.add_log_obj_ref(fwg_port_log.log_id)
+
+    def _add_log_rules_dropped(self, applied_ipt_mgrs, fwg_port_log):
+        """Add NFLOG rules to the dropped chain into iptables"""
+
+        action = log_const.DROP_EVENT
+        port_id = fwg_port_log.port_id
+        prefix = self._get_prefix(port_id, action)
+        if not prefix:
+            # Generate a new prefix from port and action
+            project_id = fwg_port_log.project_id
+            prefix = LogPrefix(port_id, action, project_id)
+            self._add_to_prefixes_table(port_id, prefix)
+            device = self._get_intf_name(port_id)
+
+            # Get iptables manager from router port
+            ipt_mgr = self._get_ipt_mgr_by_port(port_id)
+            if ipt_mgr:
+                applied_ipt_mgrs.add(ipt_mgr)
+
+            # Add the NFLOG rules to the dropped chain into iptables
+            ipv4_rules, ipv6_rules = \
+                self._generate_nflog_rules_v4v6(device, prefix=prefix.id)
+            self._add_rules_to_chain_v4v6(ipt_mgr,
+                                          'dropped', ipv4_rules, ipv6_rules,
+                                          wrap=True, top=True, tag=prefix.id)
+            # Add the NFLOG rules to the rejected chain in iptables
+            self._add_rules_to_chain_v4v6(ipt_mgr,
+                                          'rejected', ipv4_rules, ipv6_rules,
+                                          wrap=True, top=True, tag=prefix.id)
+        prefix.add_log_obj_ref(fwg_port_log.log_id)
+
+    def _add_rules_to_chain_v4v6(self, ipt_mgr, chain_name, v4_rules, v6_rules,
+                                 wrap=False, comment=None,
+                                 top=False, tag=None):
+        """Add rules to filter table in iptables and ip6tables"""
+
+        for rule in v4_rules:
+            ipt_mgr.ipv4['filter'].add_rule(chain_name, rule, wrap=wrap,
+                                            comment=comment, top=top, tag=tag)
+        for rule in v6_rules:
+            ipt_mgr.ipv6['filter'].add_rule(chain_name, rule, wrap=wrap,
+                                            comment=comment, top=top, tag=tag)
+
+    def _add_fwg_port_log(self, log_id, port_id, log_info):
+
+        self.fwg_port_logs[log_id].add(FWGPortLog(port_id, log_info))
+
+        # Add log ID into the corresponding LogPrefix object
+        if log_info['event'] == log_const.ALL_EVENT:
+            events = [log_const.ACCEPT_EVENT, log_const.DROP_EVENT]
+        else:
+            events = [log_info['event']]
+        for event in events:
+            prefix = self._get_prefix(port_id, event)
+            if prefix:
+                prefix.add_log_obj_ref(log_id)
+
+    def _generate_nflog_rules_v4v6(self, device, prefix):
+        iptables_rules = []
+        added_rules = set()
+        for direction in constants.VALID_DIRECTIONS:
+            args = self._generate_iptables_args(direction, device, prefix)
+            rule = ' '.join(args)
+            if rule in added_rules:
+                # Since these rules are already added to iptables,
+                # so we ignore them here
+                continue
+            added_rules.add(rule)
+            iptables_rules.append(rule)
+        LOG.debug("iptables-nflog-rules %r", iptables_rules)
+        return iptables_rules, iptables_rules
+
+    def _generate_iptables_args(self, direction, device, prefix=None):
+
+        direction_config = ['-%s %s' %
+                  (IPTABLES_DIRECTION_DEVICE[direction], device)]
+        match_rule = []
+        if self.rate_limit:
+            match_rule += ['-m', 'limit', '--limit', '%s/s' % self.rate_limit]
+            if self.burst_limit:
+                match_rule += ['--limit-burst %s' % self.burst_limit]
+        target = ['-j', 'NFLOG']
+        if prefix:
+            target += ['--nflog-prefix', '%s' % prefix]
+
+        args = direction_config + match_rule + target
+        return args
+
+    def _clear_rules_from_tag_v4v6(self, ipt_mgt, tag):
+        """Remove rules from filter table in iptables and ip6tables"""
+        ipt_mgt.ipv4['filter'].clear_rules_by_tag(tag)
+        ipt_mgt.ipv6['filter'].clear_rules_by_tag(tag)

neutron_fwaas/tests/unit/services/logapi/agents/drivers/iptables/test_driver.py

@@ -0,0 +1,53 @@
+# Copyright (c) 2018 Fujitsu Limited
+# 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.services.logapi.drivers import base as log_base_driver
+from neutron_fwaas.tests import base
+
+SUPPORTED_LOGGING_TYPES = ['firewall_group']
+
+
+class FakeDriver(log_base_driver.DriverBase):
+
+    @staticmethod
+    def create():
+        return FakeDriver(
+            name='fake_driver',
+            vif_types=[],
+            vnic_types=[],
+            supported_logging_types=SUPPORTED_LOGGING_TYPES,
+            requires_rpc=True
+        )
+
+
+class TestDriverBase(base.BaseTestCase):
+
+    def setUp(self):
+        super(TestDriverBase, self).setUp()
+        self.driver = FakeDriver.create()
+
+    def test_is_vif_type_compatible(self):
+        self.assertFalse(
+            self.driver.is_vif_type_compatible([]))
+
+    def test_is_vnic_compatible(self):
+        self.assertFalse(
+            self.driver.is_vnic_compatible([]))
+
+    def test_is_logging_type_supported(self):
+        self.assertTrue(
+            self.driver.is_logging_type_supported('firewall_group'))
+        self.assertFalse(
+            self.driver.is_logging_type_supported('security_group'))

neutron_fwaas/tests/unit/services/logapi/agents/drivers/iptables/test_log.py

@@ -0,0 +1,312 @@
+# Copyright (c) 2018 Fujitsu Limited.
+# 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 collections import defaultdict
+
+import mock
+from neutron.services.logapi.common import constants as log_const
+from neutron.tests.unit.api.v2 import test_base
+
+from neutron_fwaas.privileged.netfilter_log import libnetfilter_log as libnflog
+from neutron_fwaas.services.logapi.agents.drivers.iptables import log
+from neutron_fwaas.tests import base
+
+FAKE_PROJECT_ID = 'fake_project_id'
+FAKE_PORT_ID = 'fake_port_id'
+FAKE_FWG_ID = 'fake_fwg_id'
+FAKE_LOG_ID = 'fake_log_id'
+FAKE_RESOUCE_TYPE = 'firewall_group'
+
+FAKE_RATE = 100
+FAKE_BURST = 25
+
+
+class TestLogPrefix(base.BaseTestCase):
+
+    def setUp(self):
+        super(TestLogPrefix, self).setUp()
+        self.log_prefix = log.LogPrefix(FAKE_PORT_ID,
+                                        'fake_event',
+                                        FAKE_PROJECT_ID)
+        self.log_prefix.log_object_refs = set([FAKE_LOG_ID])
+
+    def test_add_log_obj_ref(self):
+        added_log_id = test_base._uuid
+        expected_log_obj_ref = set([FAKE_LOG_ID, added_log_id])
+        self.log_prefix.add_log_obj_ref(added_log_id)
+        self.assertEqual(expected_log_obj_ref, self.log_prefix.log_object_refs)
+
+    def test_remove_log_obj_ref(self):
+        expected_log_obj_ref = set()
+        self.log_prefix.remove_log_obj_ref(FAKE_LOG_ID)
+        self.assertEqual(expected_log_obj_ref, self.log_prefix.log_object_refs)
+
+    def test_is_empty(self):
+        self.log_prefix.remove_log_obj_ref(FAKE_LOG_ID)
+        result = self.log_prefix.is_empty
+        self.assertEqual(True, result)
+
+
+class BaseIptablesLogTestCase(base.BaseTestCase):
+
+    def setUp(self):
+        super(BaseIptablesLogTestCase, self).setUp()
+        self.iptables_manager_patch = mock.patch(
+            'neutron.agent.linux.iptables_manager.IptablesManager')
+        self.iptables_manager_mock = self.iptables_manager_patch.start()
+        resource_rpc_mock = mock.Mock()
+
+        self.iptables_mock = mock.Mock()
+        self.v4filter_mock = mock.Mock()
+        self.v6filter_mock = mock.Mock()
+        self.iptables_mock.ipv4 = {'filter': self.v4filter_mock}
+        self.iptables_mock.ipv6 = {'filter': self.v6filter_mock}
+
+        self.log_driver = log.IptablesLoggingDriver(mock.Mock())
+        self.log_driver.iptables_manager = self.iptables_mock
+        self.log_driver.resource_rpc = resource_rpc_mock
+        self.context = mock.Mock()
+        self.log_driver.agent_api = mock.Mock()
+
+    def test_start_logging(self):
+        fake_router_info = mock.Mock()
+        fake_router_info.router_id = 'fake_router_id'
+        fake_router_info.ns_name = 'fake_namespace'
+        libnflog.run_nflog = mock.Mock()
+        self.log_driver._create_firewall_group_log = mock.Mock()
+
+        # Test with router_info that has internal ports
+        fake_router_info.internal_ports = [
+            {'id': 'fake_port1'},
+            {'id': 'fake_port2'},
+        ]
+        fake_kwargs = {
+            'router_info': fake_router_info
+        }
+        self.log_driver.ports_belong_router = defaultdict(set)
+        self.log_driver.start_logging(self.context, **fake_kwargs)
+        self.log_driver._create_firewall_group_log.\
+            assert_called_once_with(self.context,
+                                    FAKE_RESOUCE_TYPE,
+                                    ports=fake_router_info.internal_ports,
+                                    router_id=fake_router_info.router_id)
+
+        # Test with log_resources
+        fake_kwargs = {
+            'log_resources': 'fake'
+        }
+        self.log_driver._create_firewall_group_log.reset_mock()
+        self.log_driver.start_logging(self.context, **fake_kwargs)
+        self.log_driver._create_firewall_group_log. \
+            assert_called_once_with(self.context,
+                                    FAKE_RESOUCE_TYPE,
+                                    **fake_kwargs)
+
+    def test_stop_logging(self):
+        fake_kwargs = {
+            'log_resources': 'fake'
+        }
+        self.log_driver._delete_firewall_group_log = mock.Mock()
+        self.log_driver.stop_logging(self.context, **fake_kwargs)
+        self.log_driver._delete_firewall_group_log.\
+            assert_called_once_with(self.context, **fake_kwargs)
+        fake_kwargs = {
+            'fake': 'fake'
+        }
+        self.log_driver._delete_firewall_group_log.reset_mock()
+        self.log_driver.stop_logging(self.context, **fake_kwargs)
+        self.log_driver._delete_firewall_group_log.assert_not_called()
+
+    def test_get_intf_name(self):
+        fake_router = mock.Mock()
+        fake_port_id = 'fake_router_port_id'
+
+        # Test with legacy router
+        self.log_driver.conf.agent_mode = 'legacy'
+        fake_router.router = {
+            'fake': 'fake_mode'
+        }
+        with mock.patch.object(self.log_driver.agent_api,
+                               'get_router_hosting_port',
+                               return_value=fake_router):
+            intf_name = self.log_driver._get_intf_name(fake_port_id)
+            expected_name = 'qr-fake_router'
+            self.assertEqual(expected_name, intf_name)
+
+        # Test with dvr router
+        self.log_driver.conf.agent_mode = 'dvr_snat'
+        fake_router.router = {
+            'distributed': 'fake_mode'
+        }
+        with mock.patch.object(self.log_driver.agent_api,
+                               'get_router_hosting_port',
+                               return_value=fake_router):
+            intf_name = self.log_driver._get_intf_name(fake_port_id)
+            expected_name = 'sg-fake_router'
+            self.assertEqual(expected_name, intf_name)
+
+        # Test with fip dev
+        self.log_driver.conf.agent_mode = 'dvr_snat'
+        fake_router.router = {
+            'distributed': 'fake_mode'
+        }
+        fake_router.rtr_fip_connect = 'fake'
+        self.log_driver.conf.agent_mode = 'fake'
+        with mock.patch.object(self.log_driver.agent_api,
+                               'get_router_hosting_port',
+                               return_value=fake_router):
+            intf_name = self.log_driver._get_intf_name(fake_port_id)
+            expected_name = 'rfp-fake_route'
+            self.assertEqual(expected_name, intf_name)
+
+    def test_setup_chains(self):
+        self.log_driver._add_nflog_rules_accepted = mock.Mock()
+        self.log_driver._add_log_rules_dropped = mock.Mock()
+        m_ipt_mgr = mock.Mock()
+        m_fwg_port_log = mock.Mock()
+
+        # Test with ALL event
+        m_fwg_port_log.event = log_const.ALL_EVENT
+        self.log_driver._setup_chains(m_ipt_mgr, m_fwg_port_log)
+
+        self.log_driver._add_nflog_rules_accepted.\
+            assert_called_once_with(m_ipt_mgr, m_fwg_port_log)
+        self.log_driver._add_log_rules_dropped.\
+            assert_called_once_with(m_ipt_mgr, m_fwg_port_log)
+
+        # Test with ACCEPT event
+        self.log_driver._add_nflog_rules_accepted.reset_mock()
+        self.log_driver._add_log_rules_dropped.reset_mock()
+
+        m_fwg_port_log.event = log_const.ACCEPT_EVENT
+        self.log_driver._setup_chains(m_ipt_mgr, m_fwg_port_log)
+
+        self.log_driver._add_nflog_rules_accepted.\
+            assert_called_once_with(m_ipt_mgr, m_fwg_port_log)
+        self.log_driver._add_log_rules_dropped.assert_not_called()
+
+        # Test with DROP event
+        self.log_driver._add_nflog_rules_accepted.reset_mock()
+        self.log_driver._add_log_rules_dropped.reset_mock()
+
+        m_fwg_port_log.event = log_const.DROP_EVENT
+        self.log_driver._setup_chains(m_ipt_mgr, m_fwg_port_log)
+
+        self.log_driver._add_nflog_rules_accepted.assert_not_called()
+        self.log_driver._add_log_rules_dropped.\
+            assert_called_once_with(m_ipt_mgr, m_fwg_port_log)
+
+    def test_add_nflog_rules_accepted(self):
+        ipt_mgr = mock.Mock()
+        f_accept_prefix = log.LogPrefix(FAKE_PORT_ID, log_const.
+                                        ACCEPT_EVENT,
+                                        FAKE_PROJECT_ID)
+
+        f_port_log = self._fake_port_log('fake_log_id',
+                                         log_const.ACCEPT_EVENT,
+                                         FAKE_PORT_ID)
+
+        self.log_driver._add_rules_to_chain_v4v6 = mock.Mock()
+        self.log_driver._get_ipt_mgr_by_port = mock.Mock(return_value=ipt_mgr)
+        self.log_driver._get_intf_name = mock.Mock(return_value='fake_device')
+
+        with mock.patch.object(self.log_driver, '_get_prefix',
+                               side_effect=[f_accept_prefix, None]):
+
+            # Test with prefix already added into prefixes_table
+            self.log_driver._add_nflog_rules_accepted(ipt_mgr, f_port_log)
+            self.log_driver._add_rules_to_chain_v4v6.assert_not_called()
+            self.assertEqual(set(['fake_log_id']),
+                             f_accept_prefix.log_object_refs)
+
+            # Test with prefixes_tables does not include the prefix
+            prefix = log.LogPrefix(FAKE_PORT_ID, log_const.
+                                   ACCEPT_EVENT, FAKE_PROJECT_ID)
+            with mock.patch.object(log, 'LogPrefix', return_value=prefix):
+                self.log_driver._add_nflog_rules_accepted(ipt_mgr, f_port_log)
+                v4_rules, v6_rules = self._fake_nflog_rule_v4v6('fake_device',
+                                                                prefix.id)
+
+                self.log_driver._add_rules_to_chain_v4v6.\
+                    assert_called_once_with(ipt_mgr, 'accepted',
+                                            v4_rules, v6_rules,
+                                            wrap=True, top=True, tag=prefix.id)
+                self.assertEqual(set(['fake_log_id']),
+                                 prefix.log_object_refs)
+
+    def test_add_nflog_rules_dropped(self):
+        ipt_mgr = mock.Mock()
+        f_drop_prefix = log.LogPrefix(FAKE_PORT_ID, log_const.
+                                      DROP_EVENT,
+                                      FAKE_PROJECT_ID)
+
+        f_port_log = self._fake_port_log('fake_log_id',
+                                         log_const.DROP_EVENT,
+                                         FAKE_PORT_ID)
+
+        self.log_driver._add_rules_to_chain_v4v6 = mock.Mock()
+        self.log_driver._get_ipt_mgr_by_port = mock.Mock(return_value=ipt_mgr)
+        self.log_driver._get_intf_name = mock.Mock(return_value='fake_device')
+
+        with mock.patch.object(self.log_driver, '_get_prefix',
+                               side_effect=[f_drop_prefix, None]):
+
+            # Test with prefix already added into prefixes_table
+            self.log_driver._add_log_rules_dropped(ipt_mgr, f_port_log)
+            self.log_driver._add_rules_to_chain_v4v6.assert_not_called()
+            self.assertEqual(set(['fake_log_id']),
+                             f_drop_prefix.log_object_refs)
+
+            # Test with prefixes_tables does not include the prefix
+            prefix = log.LogPrefix(FAKE_PORT_ID, log_const.
+                                   ACCEPT_EVENT, FAKE_PROJECT_ID)
+            with mock.patch.object(log, 'LogPrefix', return_value=prefix):
+                self.log_driver._add_log_rules_dropped(ipt_mgr, f_port_log)
+                v4_rules, v6_rules = self._fake_nflog_rule_v4v6('fake_device',
+                                                                prefix.id)
+
+                calls = [
+                    mock.call(ipt_mgr, 'dropped', v4_rules, v6_rules,
+                              wrap=True, top=True, tag=prefix.id),
+                    mock.call(ipt_mgr, 'rejected', v4_rules, v6_rules,
+                              wrap=True, top=True, tag=prefix.id),
+                ]
+                self.log_driver._add_rules_to_chain_v4v6.\
+                    assert_has_calls(calls)
+                self.assertEqual(set(['fake_log_id']),
+                                 prefix.log_object_refs)
+
+    def _fake_port_log(self, log_id, event, port_id):
+        f_log_info = {
+            'event': event,
+            'project_id': FAKE_PROJECT_ID,
+            'id': log_id
+        }
+        return log.FWGPortLog(port_id, f_log_info)
+
+    def _fake_nflog_rule_v4v6(self, device, tag):
+        v4_nflog_rule = ['-i %s -m limit --limit %s/s --limit-burst %s '
+                         '-j NFLOG --nflog-prefix %s'
+                         % (device, FAKE_RATE, FAKE_BURST, tag)]
+        v4_nflog_rule += ['-o %s -m limit --limit %s/s --limit-burst %s '
+                         '-j NFLOG --nflog-prefix %s'
+                         % (device, FAKE_RATE, FAKE_BURST, tag)]
+        v6_nflog_rule = ['-i %s -m limit --limit %s/s --limit-burst %s '
+                         '-j NFLOG --nflog-prefix %s'
+                         % (device, FAKE_RATE, FAKE_BURST, tag)]
+        v6_nflog_rule += ['-o %s -m limit --limit %s/s --limit-burst %s '
+                          '-j NFLOG --nflog-prefix %s'
+                          % (device, FAKE_RATE, FAKE_BURST, tag)]
+        return v4_nflog_rule, v6_nflog_rule

neutron_fwaas/tests/unit/services/logapi/base.py

@@ -0,0 +1,38 @@
+#    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 neutron.api.rpc.callbacks.consumer import registry as cons_registry
+from neutron.api.rpc.callbacks.producer import registry as prod_registry
+from neutron.api.rpc.callbacks import resource_manager
+from neutron.tests.unit import testlib_api
+
+
+class BaseLogTestCase(testlib_api.SqlTestCase):
+    def setUp(self):
+        super(BaseLogTestCase, self).setUp()
+
+        with mock.patch.object(
+            resource_manager.ResourceCallbacksManager, '_singleton',
+            new_callable=mock.PropertyMock(return_value=False)):
+
+            self.cons_mgr = resource_manager.ConsumerResourceCallbacksManager()
+            self.prod_mgr = resource_manager.ProducerResourceCallbacksManager()
+            for mgr in (self.cons_mgr, self.prod_mgr):
+                mgr.clear()
+
+        mock.patch.object(
+            cons_registry, '_get_manager', return_value=self.cons_mgr).start()
+
+        mock.patch.object(
+            prod_registry, '_get_manager', return_value=self.prod_mgr).start()

setup.cfg

@@ -61,6 +61,8 @@ neutron.agent.l3.extensions =
 neutron.agent.l3.firewall_drivers =
     conntrack = neutron_fwaas.services.firewall.service_drivers.agents.drivers.linux.legacy_conntrack:ConntrackLegacy
     netlink_conntrack = neutron_fwaas.services.firewall.service_drivers.agents.drivers.linux.netlink_conntrack:ConntrackNetlink
+neutron.services.logapi.drivers =
+    fwaas_v2_log = neutron_fwaas.services.logapi.agents.drivers.iptables.log:IptablesLoggingDriver
 
 [extract_messages]
 keywords = _ gettext ngettext l_ lazy_gettext