Merge branch 'feature-firewall-fixes' into 'master'

Feature Firewall Fixes

circle/firewall/fw.py

@@ -319,7 +319,8 @@ def dhcp():
 
 
 def vlan():
-    obj = Vlan.objects.values('vid', 'name', 'network4', 'network6')
+    obj = Vlan.objects.filter(managed=True).values(
+        'vid', 'name', 'network4', 'network6')
     retval = {x['name']: {'tag': x['vid'],
                           'type': 'internal',
                           'interfaces': [x['name']],

circle/firewall/models.py

@@ -35,10 +35,12 @@ import django.conf
 from django.db.models.signals import post_save, post_delete
 import random
 
-from common.models import HumanSortField
+from common.models import method_cache, WorkerNotFound, HumanSortField
 from firewall.tasks.local_tasks import reloadtask
 from .iptables import IptRule
 from acl.models import AclBase
+
+
 logger = logging.getLogger(__name__)
 settings = django.conf.settings.FIREWALL_SETTINGS
 
@@ -189,19 +191,31 @@ class Rule(models.Model):
     def get_absolute_url(self):
         return ('network.rule', None, {'pk': self.pk})
 
-    @staticmethod
-    def get_chain_name(local, remote, direction):
-        if direction == 'in':
+    def get_chain_name(self, local, remote):
+        if local:  # host or vlan
+            if self.direction == 'in':
                 # remote -> local
-            return '%s_%s' % (remote, local)
+                return '%s_%s' % (remote.name, local.name)
             else:
                 # local -> remote
-            return '%s_%s' % (local, remote)
+                return '%s_%s' % (local.name, remote.name)
+            # firewall rule
+        elif self.firewall_id:
+            return 'INPUT' if self.direction == 'in' else 'OUTPUT'
+
+    def get_dport_sport(self):
+        if self.direction == 'in':
+            return self.dport, self.sport
+        else:
+            return self.sport, self.dport
 
     def get_ipt_rules(self, host=None):
         # action
         action = 'LOG_ACC' if self.action == 'accept' else 'LOG_DROP'
 
+        # 'chain_name': rule dict
+        retval = {}
+
         # src and dst addresses
         src = None
         dst = None
@@ -212,34 +226,28 @@ class Rule(models.Model):
                 dst = ip
             else:
                 src = ip
-
-        # src and dst ports
-        if self.direction == 'in':
-            dport = self.dport
-            sport = self.sport
+            vlan = host.vlan
+        elif self.vlan_id:
+            vlan = self.vlan
         else:
-            dport = self.sport
-            sport = self.dport
+            vlan = None
 
-        # 'chain_name': rule dict
-        retval = {}
+        if vlan and not vlan.managed:
+            return retval
+
+        # src and dst ports
+        dport, sport = self.get_dport_sport()
 
         # process foreign vlans
         for foreign_vlan in self.foreign_network.vlans.all():
+            if not foreign_vlan.managed:
+                continue
+
             r = IptRule(priority=self.weight, action=action,
                         proto=self.proto, extra=self.extra,
                         comment='Rule #%s' % self.pk,
                         src=src, dst=dst, dport=dport, sport=sport)
-            # host, hostgroup or vlan rule
-            if host or self.vlan_id:
-                local_vlan = host.vlan.name if host else self.vlan.name
-                chain_name = Rule.get_chain_name(local=local_vlan,
-                                                 remote=foreign_vlan.name,
-                                                 direction=self.direction)
-            # firewall rule
-            elif self.firewall_id:
-                chain_name = 'INPUT' if self.direction == 'in' else 'OUTPUT'
-
+            chain_name = self.get_chain_name(local=vlan, remote=foreign_vlan)
             retval[chain_name] = r
 
         return retval
@@ -796,6 +804,20 @@ class Firewall(models.Model):
     def __unicode__(self):
         return self.name
 
+    @method_cache(30)
+    def get_remote_queue_name(self, queue_id):
+        """Returns the name of the remote celery queue for this node.
+
+        Throws Exception if there is no worker on the queue.
+        The result may include dead queues because of caching.
+        """
+        from .tasks.remote_tasks import check_queue
+
+        if check_queue(self.name, queue_id, None):
+            return self.name + "." + queue_id
+        else:
+            raise WorkerNotFound()
+
 
 class Domain(models.Model):
     name = models.CharField(max_length=40, validators=[val_domain],

circle/firewall/tasks/local_tasks.py

@@ -20,8 +20,10 @@ from socket import gethostname
 
 import django.conf
 from django.core.cache import cache
+from celery.exceptions import TimeoutError
 
 from manager.mancelery import celery
+from common.models import WorkerNotFound
 
 settings = django.conf.settings.FIREWALL_SETTINGS
 logger = getLogger(__name__)
@@ -36,19 +38,37 @@ def _apply_once(name, queues, task, data):
         return
     cache.delete(lockname)
 
+    data = data()
     for queue in queues:
-        task.apply_async(args=data(), queue=queue)
-    logger.info("%s configuration is reloaded.", name)
+        try:
+            task.apply_async(args=data, queue=queue, expires=60).get(timeout=5)
+            logger.info("%s configuration is reloaded. (queue: %s)",
+                        name, queue)
+        except TimeoutError as e:
+            logger.critical('%s (queue: %s)', e, queue)
+        except:
+            logger.critical('Unhandled exception: queue: %s data: %s',
+                            queue, data, exc_info=True)
+
+
+def get_firewall_queues():
+    from firewall.models import Firewall
+    retval = []
+    for fw in Firewall.objects.all():
+        try:
+            retval.append(fw.get_remote_queue_name('firewall'))
+        except WorkerNotFound:
+            logger.critical('Firewall %s is offline', fw.name)
+    return list(retval)
 
 
 @celery.task(ignore_result=True)
-def periodic_task():
+def reloadtask_worker():
     from firewall.fw import BuildFirewall, dhcp, dns, ipset, vlan
     from remote_tasks import (reload_dns, reload_dhcp, reload_firewall,
                               reload_firewall_vlan, reload_blacklist)
 
-    firewall_queues = [("%s.firewall" % i) for i in
-                       settings.get('firewall_queues', [gethostname()])]
+    firewall_queues = get_firewall_queues()
     dns_queues = [("%s.dns" % i) for i in
                   settings.get('dns_queues', [gethostname()])]
 
@@ -78,5 +98,5 @@ def reloadtask(type='Host', timeout=15):
     }[type]
     logger.info("Reload %s on next periodic iteration applying change to %s.",
                 ", ".join(reload), type)
-    for i in reload:
-        cache.add("%s_lock" % i, "true", 30)
+    if all(cache.add("%s_lock" % i, True, 30) for i in reload):
+        reloadtask_worker.apply_async(queue='localhost.man', countdown=5)

circle/firewall/tasks/remote_tasks.py

@@ -18,6 +18,24 @@
 from manager.mancelery import celery
 
 
+def check_queue(firewall, queue_id, priority):
+    ''' Celery inspect job to check for active workers at queue_id
+        return True/False
+    '''
+    queue_name = firewall + "." + queue_id
+    if priority is not None:
+        queue_name = queue_name + "." + priority
+    inspect = celery.control.inspect()
+    inspect.timeout = 0.1
+    active_queues = inspect.active_queues()
+    if active_queues is None:
+        return False
+
+    queue_names = (queue['name'] for worker in active_queues.itervalues()
+                   for queue in worker)
+    return queue_name in queue_names
+
+
 @celery.task(name='firewall.reload_dns')
 def reload_dns(data):
     pass

circle/firewall/templates/firewall/iptables.conf

@@ -27,13 +27,21 @@ COMMIT
 -A LOG_ACC -j ACCEPT
 
 # initialize FORWARD chain
+{% if proto == "ipv4" %}
 -A FORWARD -m set --match-set blacklist src,dst -j DROP
+{% endif %}
 -A FORWARD -m state --state INVALID -g LOG_DROP
 -A FORWARD -m state --state ESTABLISHED,RELATED -j ACCEPT
+{% if proto == "ipv4" %}
 -A FORWARD -p icmp --icmp-type echo-request -g LOG_ACC
+{% else %}
+-A FORWARD -p icmpv6 --icmpv6-type echo-request -g LOG_ACC
+{% endif %}
 
 # initialize INPUT chain
+{% if proto == "ipv4" %}
 -A INPUT -m set --match-set blacklist src -j DROP
+{% endif %}
 -A INPUT -m state --state INVALID -g LOG_DROP
 -A INPUT -i lo -j ACCEPT
 -A INPUT -m state --state ESTABLISHED,RELATED -j ACCEPT

circle/firewall/tests/test_firewall.py

@@ -23,7 +23,7 @@ from ..admin import HostAdmin
 from firewall.models import (Vlan, Domain, Record, Host, VlanGroup, Group,
                              Rule, Firewall)
 from firewall.fw import dns, ipv6_to_octal
-from firewall.tasks.local_tasks import periodic_task, reloadtask
+from firewall.tasks.local_tasks import reloadtask_worker, reloadtask
 from django.forms import ValidationError
 from ..iptables import IptRule, IptChain, InvalidRuleExcepion
 from mock import patch
@@ -323,6 +323,6 @@ class ReloadTestCase(TestCase):
         with patch('firewall.tasks.local_tasks.cache') as cache:
             self.test_host_add_port()
             self.test_host_add_port2()
-            periodic_task()
+            reloadtask_worker()
             reloadtask()
             assert cache.delete.called

circle/manager/mancelery.py

@@ -45,11 +45,6 @@ celery.conf.update(
               routing_key="monitor"),
     ),
     CELERYBEAT_SCHEDULE={
-        'firewall.periodic_task': {
-            'task': 'firewall.tasks.local_tasks.periodic_task',
-            'schedule': timedelta(seconds=5),
-            'options': {'queue': 'localhost.man'}
-        },
         'vm.update_domain_states': {
             'task': 'vm.tasks.local_periodic_tasks.update_domain_states',
             'schedule': timedelta(seconds=10),

circle/monitor/tasks/local_periodic_tasks.py

@@ -23,7 +23,9 @@ from django.conf import settings
 from manager.mancelery import celery
 
 from vm.tasks.vm_tasks import check_queue
+from firewall.tasks.remote_tasks import check_queue as check_queue_fw
 from vm.models import Node, InstanceTemplate
+from firewall.models import Firewall
 from storage.models import DataStore
 from monitor.client import Client
 
@@ -80,6 +82,11 @@ def check_celery_queues():
                                            "storage-" + s, is_queue_alive,
                                            time()))
 
+    for fw in Firewall.objects.all():
+        is_queue_alive = check_queue_fw(fw.name, "firewall", None)
+        metrics.append(graphite_string(
+            "firewall", fw.name, "firewall", is_queue_alive, time()))
+
     Client().send(metrics)