flake8: resolve flake8 violations (smicallef#1756)

modules/sfp__stor_db.py

@@ -51,11 +51,10 @@ def handleEvent(self, sfEvent):
         if not self.opts['_store']:
             return
 
-        if self.opts['maxstorage'] != 0:
-            if len(sfEvent.data) > self.opts['maxstorage']:
-                self.debug("Storing an event: " + sfEvent.eventType)
-                self.__sfdb__.scanEventStore(self.getScanId(), sfEvent, self.opts['maxstorage'])
-                return
+        if self.opts['maxstorage'] != 0 and len(sfEvent.data) > self.opts['maxstorage']:
+            self.debug("Storing an event: " + sfEvent.eventType)
+            self.__sfdb__.scanEventStore(self.getScanId(), sfEvent, self.opts['maxstorage'])
+            return
 
         self.debug("Storing an event: " + sfEvent.eventType)
         self.__sfdb__.scanEventStore(self.getScanId(), sfEvent)

modules/sfp__stor_stdout.py

@@ -85,10 +85,11 @@ def output(self, event):
             srcdata = srcdata[0:self.opts['_maxlength']]
 
         if self.opts['_format'] == "tab":
+            event_type = self.opts['_eventtypes'][event.eventType]
             if self.opts['_showsource']:
-                print(('{0:30}\t{1:45}\t{2}\t{3}'.format(event.module, self.opts['_eventtypes'][event.eventType], srcdata, data)))
+                print(f"{event.module.ljust(30)}\t{event_type.ljust(45)}\t{srcdata}\t{data}")
             else:
-                print(('{0:30}\t{1:45}\t{2}'.format(event.module, self.opts['_eventtypes'][event.eventType], data)))
+                print(f"{event.module.ljust(30)}\t{event_type.ljust(45)}\t{data}")
 
         if self.opts['_format'] == "csv":
             print((event.module + d + self.opts['_eventtypes'][event.eventType] + d + srcdata + d + data))

modules/sfp_binaryedge.py

@@ -267,9 +267,8 @@ def handleEvent(self, event):
                         continue
 
                     if self.getTarget().matches(host, includeParents=True):
-                        if self.opts['verify']:
-                            if not self.sf.resolveHost(host) and not self.sf.resolveHost6(host):
-                                continue
+                        if self.opts['verify'] and not self.sf.resolveHost(host) and not self.sf.resolveHost6(host):
+                            continue
 
                         evt = SpiderFootEvent("INTERNET_NAME", host, self.__name__, event)
                         self.notifyListeners(evt)
@@ -303,10 +302,9 @@ def handleEvent(self, event):
 
                     self.reportedhosts[rec] = True
 
-                    if self.opts['verify']:
-                        if not self.sf.resolveHost(rec) and not self.sf.resolveHost6(rec):
-                            self.debug(f"Couldn't resolve {rec}, so skipping.")
-                            continue
+                    if self.opts['verify'] and not self.sf.resolveHost(rec) and not self.sf.resolveHost6(rec):
+                        self.debug(f"Couldn't resolve {rec}, so skipping.")
+                        continue
 
                     e = SpiderFootEvent('INTERNET_NAME', rec, self.__name__, event)
                     self.notifyListeners(e)

modules/sfp_dnsdumpster.py

@@ -79,8 +79,8 @@ def query(self, domain):
             self.error("Error obtaining CSRF tokens")
             self.errorState = True
             return ret
-        else:
-            self.debug("Successfully obtained CSRF tokens")
+
+        self.debug("Successfully obtained CSRF tokens")
 
         # Otherwise, do the needful
         url = "https://dnsdumpster.com/"

modules/sfp_greynoise.py

@@ -165,28 +165,20 @@ def handleEvent(self, event):
         if eventName == "NETBLOCK_OWNER":
             if not self.opts["netblocklookup"]:
                 return
-            else:
-                if IPNetwork(eventData).prefixlen < self.opts["maxnetblock"]:
-                    self.debug(
-                        "Network size bigger than permitted: "
-                        + str(IPNetwork(eventData).prefixlen)
-                        + " > "
-                        + str(self.opts["maxnetblock"])
-                    )
-                    return
+
+            max_netblock = self.opts['maxnetblock']
+            if IPNetwork(eventData).prefixlen < max_netblock:
+                self.debug(f"Network size bigger than permitted: {IPNetwork(eventData).prefixlen} > {max_netblock}")
+                return
 
         if eventName == "NETBLOCK_MEMBER":
             if not self.opts["subnetlookup"]:
                 return
-            else:
-                if IPNetwork(eventData).prefixlen < self.opts["maxsubnet"]:
-                    self.debug(
-                        "Network size bigger than permitted: "
-                        + str(IPNetwork(eventData).prefixlen)
-                        + " > "
-                        + str(self.opts["maxsubnet"])
-                    )
-                    return
+
+            max_subnet = self.opts['maxsubnet']
+            if IPNetwork(eventData).prefixlen < max_subnet:
+                self.debug(f"Network size bigger than permitted: {IPNetwork(eventData).prefixlen} > {max_subnet}")
+                return
 
         if eventName == "IP_ADDRESS":
             evtType = "MALICIOUS_IPADDR"

modules/sfp_hosting.py

@@ -62,11 +62,12 @@ def queryAddr(self, qaddr):
         data['content'] = self.sf.cacheGet("sfipcat", 48)
         if data['content'] is None:
             data = self.sf.fetchUrl(url, useragent=self.opts['_useragent'])
+
             if data['content'] is None:
                 self.error("Unable to fetch " + url)
                 return None
-            else:
-                self.sf.cachePut("sfipcat", data['content'])
+
+            self.sf.cachePut("sfipcat", data['content'])
 
         for line in data['content'].split('\n'):
             if "," not in line:

modules/sfp_onyphe.py

@@ -133,7 +133,8 @@ def query(self, endpoint, ip, page=1):
                 )
                 self.errorState = True
                 return None
-            elif "results" not in info or info["results"] == []:
+
+            if "results" not in info or info["results"] == []:
                 self.info(f"No Onyphe {endpoint} data found for {ip}")
                 return None
         except Exception as e:
@@ -213,12 +214,9 @@ def emitDomainData(self, response, eventData, event):
                     self.debug("Host no longer resolves to our IP.")
                     continue
 
-                if not self.opts["cohostsamedomain"]:
-                    if self.getTarget().matches(domain, includeParents=True):
-                        self.debug(
-                            "Skipping " + domain + " because it is on the same domain."
-                        )
-                        continue
+                if not self.opts["cohostsamedomain"] and self.getTarget().matches(domain, includeParents=True):
+                    self.debug(f"Skipping {domain} because it is on the same domain.")
+                    continue
 
                 evt = SpiderFootEvent("CO_HOSTED_SITE", domain, self.__name__, event)
                 self.notifyListeners(evt)

modules/sfp_securitytrails.py

@@ -125,6 +125,7 @@ def query(self, qry, querytype, page=1, accum=None):
             info = json.loads(res['content'])
             if querytype == "domain":
                 return info.get('subdomains', None)
+
             if info.get("record_count", 0) > 100:
                 if len(info.get('records', [])) >= 100:
                     # Avoid throttling
@@ -134,12 +135,12 @@ def query(self, qry, querytype, page=1, accum=None):
                     else:
                         accum = info.get('records')
                     return self.query(qry, querytype, page + 1, accum)
-                else:
-                    # We are at the last page
-                    accum.extend(info.get('records', []))
-                    return accum
-            else:
-                return info.get('records', [])
+
+                # We are at the last page
+                accum.extend(info.get('records', []))
+                return accum
+
+            return info.get('records', [])
         except Exception as e:
             self.error("Error processing JSON response from SecurityTrails: " + str(e))
             return None

modules/sfp_socialprofiles.py

@@ -219,8 +219,8 @@ def handleEvent(self, event):
                     self.debug("Match found: " + match)
                     if match in instances:
                         continue
-                    else:
-                        instances.append(match)
+
+                    instances.append(match)
 
                     if self.checkForStop():
                         return
@@ -242,17 +242,18 @@ def handleEvent(self, event):
 
                         if pres["content"] is None:
                             continue
-                        else:
-                            found = False
-                            for kw in self.keywords:
-                                if re.search(
-                                    r"[^a-zA-Z\-\_]" + kw + r"[^a-zA-Z\-\_]",
-                                    pres["content"],
-                                    re.IGNORECASE,
-                                ):
-                                    found = True
-                            if not found:
-                                continue
+
+                        found = False
+                        for kw in self.keywords:
+                            if re.search(
+                                r"[^a-zA-Z\-\_]" + kw + r"[^a-zA-Z\-\_]",
+                                pres["content"],
+                                re.IGNORECASE,
+                            ):
+                                found = True
+
+                        if not found:
+                            continue
 
                     self.info("Social Media Profile found at " + site + ": " + match)
                     match = urllib.parse.unquote(match)

modules/sfp_tool_nbtscan.py

@@ -117,12 +117,12 @@ def handleEvent(self, event):
         if eventData in self.results:
             self.debug(f"Skipping {eventData} as already scanned.")
             return
-        else:
-            # Might be a subnet within a subnet or IP within a subnet
-            for addr in self.results:
-                if IPNetwork(eventData) in IPNetwork(addr):
-                    self.debug(f"Skipping {eventData} as already within a scanned range.")
-                    return
+
+        # Might be a subnet within a subnet or IP within a subnet
+        for addr in self.results:
+            if IPNetwork(eventData) in IPNetwork(addr):
+                self.debug(f"Skipping {eventData} as already within a scanned range.")
+                return
 
         self.results[eventData] = True
 

modules/sfp_tool_nmap.py

@@ -104,12 +104,12 @@ def handleEvent(self, event):
         if eventData in self.results:
             self.debug("Skipping " + eventData + " as already scanned.")
             return
-        else:
-            # Might be a subnet within a subnet or IP within a subnet
-            for addr in self.results:
-                if IPNetwork(eventData) in IPNetwork(addr):
-                    self.debug("Skipping " + eventData + " as already within a scanned range.")
-                    return
+
+        # Might be a subnet within a subnet or IP within a subnet
+        for addr in self.results:
+            if IPNetwork(eventData) in IPNetwork(addr):
+                self.debug(f"Skipping {eventData} as already within a scanned range.")
+                return
 
         self.results[eventData] = True
 

modules/sfp_tool_onesixtyone.py

@@ -66,10 +66,9 @@ def setup(self, sfc, userOpts=dict()):
         # Write communities to file for use later on
         try:
             _, self.communitiesFile = tempfile.mkstemp("communities")
-            f = open(self.communitiesFile, "w")
-            for community in self.opts['communities'].split(","):
-                f.write(community.strip() + "\n")
-            f.close()
+            with open(self.communitiesFile, "w") as f:
+                for community in self.opts['communities'].split(","):
+                    f.write(community.strip() + "\n")
         except BaseException as e:
             self.error(f"Unable to write communities file ({self.communitiesFile}): {e}")
             self.errorState = True
@@ -133,12 +132,12 @@ def handleEvent(self, event):
         if eventData in self.results:
             self.debug(f"Skipping {eventData} as already scanned.")
             return
-        else:
-            # Might be a subnet within a subnet or IP within a subnet
-            for addr in self.results:
-                if IPNetwork(eventData) in IPNetwork(addr):
-                    self.debug(f"Skipping {eventData} as already within a scanned range.")
-                    return
+
+        # Might be a subnet within a subnet or IP within a subnet
+        for addr in self.results:
+            if IPNetwork(eventData) in IPNetwork(addr):
+                self.debug(f"Skipping {eventData} as already within a scanned range.")
+                return
 
         self.results[eventData] = True
 

modules/sfp_tool_testsslsh.py

@@ -133,17 +133,17 @@ def handleEvent(self, event):
         if eventData in self.results:
             self.debug(f"Skipping {eventData} as already scanned.")
             return
-        else:
-            if eventName != "INTERNET_NAME":
-                # Might be a subnet within a subnet or IP within a subnet
-                for addr in self.results:
-                    try:
-                        if IPNetwork(eventData) in IPNetwork(addr):
-                            self.debug(f"Skipping {eventData} as already within a scanned range.")
-                            return
-                    except BaseException:
-                        # self.results will also contain hostnames
-                        continue
+
+        if eventName != "INTERNET_NAME":
+            # Might be a subnet within a subnet or IP within a subnet
+            for addr in self.results:
+                try:
+                    if IPNetwork(eventData) in IPNetwork(addr):
+                        self.debug(f"Skipping {eventData} as already within a scanned range.")
+                        return
+                except BaseException:
+                    # self.results will also contain hostnames
+                    continue
 
         # If we weren't passed a netblock, this will be empty
         if not targets:

modules/sfp_viewdns.py

@@ -240,10 +240,9 @@ def handleEvent(self, event):
                 if self.cohostcount >= self.opts['maxcohost']:
                     continue
 
-                if eventName in ["IP_ADDRESS", "IPV6_ADDRESS"] and self.opts['verify']:
-                    if not self.sf.validateIP(domain, eventData):
-                        self.debug(f"Host {domain} no longer resolves to IP address: {eventData}")
-                        continue
+                if eventName in ["IP_ADDRESS", "IPV6_ADDRESS"] and self.opts['verify'] and not self.sf.validateIP(domain, eventData):
+                    self.debug(f"Host {domain} no longer resolves to IP address: {eventData}")
+                    continue
 
                 self.cohostcount += 1
 

modules/sfp_whoxy.py

@@ -101,23 +101,25 @@ def query(self, qry, querytype, page=1, accum=None):
 
         try:
             info = json.loads(res['content'])
+
             if info.get("status", 0) == 0:
                 self.error("Error querying Whoxy: " + info.get("status_reason", "Unknown"))
                 self.errorState = True
                 return None
+
             if info.get("total_pages", 1) > 1:
                 if info.get("current_page") < info.get("total_pages"):
                     if accum:
                         accum.extend(info.get('search_result'))
                     else:
                         accum = info.get('search_result')
                     return self.query(qry, querytype, page + 1, accum)
-                else:
-                    # We are at the last page
-                    accum.extend(info.get('search_result', []))
-                    return accum
-            else:
-                return info.get('search_result', [])
+
+                # We are at the last page
+                accum.extend(info.get('search_result', []))
+                return accum
+
+            return info.get('search_result', [])
         except Exception as e:
             self.error("Error processing JSON response from Whoxy: " + str(e))
             return None

setup.cfg

@@ -4,27 +4,20 @@ max-complexity = 60
 docstring-convention = google
 ignore-decorators = property
 select = C,E,F,W,B,B9,DAR,DUO,R,A,S,Q0,SIM,SFS
-# Note: ANN and SIM tests should be reviewed and fixed instead of ignored
-extend-ignore = E501 W503 B006 E800 B950 SFS301 SF01 Q000 SIM102 SIM113 SIM114 B902 B907 I D ANN
+# Note: B902, B907 and ANN should be fixed instead of ignored
+extend-ignore = E501 W503 B006 B950 SFS301 SF01 Q000 B902 B907 ANN
 # Note: most of these should be fixed instead of ignored
 per-file-ignores =
     spiderfoot/event.py:A003
-    spiderfoot/plugin.py:B902
-    spiderfoot/db.py:SFS101,B902
-    sf.py:SFS201,B902
-    sflib.py:SFS101,SIM110,B902
-    sfscan.py:SIM105,B902
-    modules/*:SIM115,B902,B907,R505
-    modules/sfp_alienvault.py:B902,C901
-    modules/sfp_binaryedge.py:B902,C901
-    modules/sfp__stor_stdout.py:SFS201
+    spiderfoot/db.py:SFS101
+    modules/*:SIM102,SIM113,SIM114
+    modules/sfp_alienvault.py:C901
+    modules/sfp_binaryedge.py:C901
     modules/sfp_bitcoin.py:SFS101
-    modules/sfp_socialprofiles.py:R507
-    spiderfoot/correlation.py:SIM110,B902,E275
     spiderfoot/__init__.py:F401
-    sfcli.py:DAR,B902
-    sfwebui.py:A001,A002,B902,B905,R505
-    test/*:SIM117,B904,E241,ANN
+    sfcli.py:DAR
+    sfwebui.py:A001,A002,B905
+    test/*:SIM117,B904,ANN
     docs/conf.py:A
 
 [darglint]