def run_legacy_checks()

in client/blocky.py [0:0]

• 61 lines of code
• 3 McCabe index (conditional complexity)

def run_legacy_checks():
   """ Runs checks using the legacy blocky UI server (mod_lua) """
   apiurl = CONFIG['server']['legacyurl']
   actions = []
   mylist = getbans()
   try:
      actions = requests.get(apiurl).json()
      syslog.syslog(syslog.LOG_INFO, "Fetched a total of %u firewall actions from %s" % (len(actions), apiurl))
   except:
      syslog.syslog(syslog.LOG_WARNING, "Could not retrieve blocky actions list from %s - server down??!" % apiurl)

   whitelist = [] # Things we are unbanning, and thus shouldn't just ban right again

   # For each action element, find out what to do, and who to do it to.
   for action in actions:

      # Unban request
      target = action.get('target', '*')
      if 'unban' in action:
         if target == '*' or target == CONFIG['client']['hostname']:
            ip = action.get('ip')
            if ip:
               ip = ip.strip()
               block = None
               if '/' in ip:
                  block = netaddr.IPNetwork(ip)
               else:
                  if ':' in ip:
                     block = netaddr.IPNetwork("%s/128" % ip) # IPv6
                  else:
                     block = netaddr.IPNetwork("%s/32" % ip)  # IPv4
               whitelist.append(block)
               found = inlist(mylist, ip)
               if found:
                  entry = found[0]
                  syslog.syslog(syslog.LOG_INFO, "Removing %s from block list (found at line %s as %s)" % (ip, entry['linenumber'], entry['source']))
                  if not unban_line(ip, found[0]['linenumber']):
                     syslog.syslog(syslog.LOG_WARNING, "Could not remove ban for %s from iptables!" % ip)
                  else:
                     mylist = getbans() # Refresh after action succeeded

      # Ban request?
      elif 'ip' in action:
         if target == '*' or target == CONFIG['client']['hostname']:
            ip = action.get('ip')
            if ip:
               ip = ip.strip() # backwards compat
               banit = True
               block = None
               if '/' in ip:
                  block = netaddr.IPNetwork(ip)
               else:
                  if ':' in ip:
                     block = netaddr.IPNetwork("%s/128" % ip) # IPv6
                  else:
                     block = netaddr.IPNetwork("%s/32" % ip)  # IPv4
               for wblock in whitelist:
                  if block in wblock or wblock in block:
                     syslog.syslog(syslog.LOG_WARNING, "%s was requested banned but %s is whitelisted, ignoring ban" % (block, wblock))
                     banit = False
               if banit:
                  found = inlist(mylist, ip)
                  if not found:
                     reason = action.get('reason', "No reason specified")
                     syslog.syslog(syslog.LOG_INFO, "Adding %s to block list; %s" % (ip, reason))
                     if not ban(ip):
                        syslog.syslog(syslog.LOG_WARNING, "Could not add ban for %s in iptables!" % ip)
                     else:
                        mylist = getbans() # Refresh after action succeeded