smicallef/spiderfoot
1
0
Fork 0
mirror of https://github.com/smicallef/spiderfoot.git synced 2026-10-12 02:22:47 +00:00
Code Issues Packages Projects Releases Wiki Activity

flake8: resolve flake8 violations (#1756)

Browse Source
This commit is contained in:
bcoles
2023-02-20 20:12:21 +01:00
committed by GitHub
parent c6d7afb640
commit 4251bd54ca
25 changed files with 169 additions and 209 deletions
Download Patch File Download Diff File Expand all files Collapse all files
modules/sfp__stor_db.py
+4 -5
View File
@@ -51,11 +51,10 @@ class sfp__stor_db(SpiderFootPlugin):
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
+3 -2
View File
@@ -85,10 +85,11 @@ class sfp__stor_stdout(SpiderFootPlugin):
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
+5 -7
View File
@@ -267,9 +267,8 @@ class sfp_binaryedge(SpiderFootPlugin):
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 @@ class sfp_binaryedge(SpiderFootPlugin):
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
+2 -2
View File
@@ -79,8 +79,8 @@ class sfp_dnsdumpster(SpiderFootPlugin):
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
+10 -18
View File
@@ -165,28 +165,20 @@ class sfp_greynoise(SpiderFootPlugin):
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
+3 -2
View File
@@ -62,11 +62,12 @@ class sfp_hosting(SpiderFootPlugin):
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
+5 -7
View File
@@ -133,7 +133,8 @@ class sfp_onyphe(SpiderFootPlugin):
)
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 @@ class sfp_onyphe(SpiderFootPlugin):
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
+7 -6
View File
@@ -125,6 +125,7 @@ class sfp_securitytrails(SpiderFootPlugin):
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 @@ class sfp_securitytrails(SpiderFootPlugin):
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
+14 -13
View File
@@ -219,8 +219,8 @@ class sfp_socialprofiles(SpiderFootPlugin):
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 @@ class sfp_socialprofiles(SpiderFootPlugin):
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
+6 -6
View File
@@ -117,12 +117,12 @@ class sfp_tool_nbtscan(SpiderFootPlugin):
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
+6 -6
View File
@@ -104,12 +104,12 @@ class sfp_tool_nmap(SpiderFootPlugin):
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
+9 -10
View File
@@ -66,10 +66,9 @@ class sfp_tool_onesixtyone(SpiderFootPlugin):
# 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 @@ class sfp_tool_onesixtyone(SpiderFootPlugin):
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
+11 -11
View File
@@ -133,17 +133,17 @@ class sfp_tool_testsslsh(SpiderFootPlugin):
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
+3 -4
View File
@@ -240,10 +240,9 @@ class sfp_viewdns(SpiderFootPlugin):
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
+8 -6
View File
@@ -101,10 +101,12 @@ class sfp_whoxy(SpiderFootPlugin):
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:
@@ -112,12 +114,12 @@ class sfp_whoxy(SpiderFootPlugin):
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
+9 -16
View File
@@ -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]
sf.py
+4 -4
View File
@@ -201,7 +201,7 @@ def main() -> None:
for m in sorted(sfModules.keys()):
if "__" in m:
continue
print(('{0:25} {1}'.format(m, sfModules[m]['descr'])))
print(f"{m.ljust(25)} {sfModules[m]['descr']}")
sys.exit(0)
if args.types:
@@ -213,7 +213,7 @@ def main() -> None:
types[r[1]] = r[0]
for t in sorted(types.keys()):
print(('{0:45} {1}'.format(t, types[t])))
print(f"{t.ljust(45)} {types[t]}")
sys.exit(0)
if args.l:
@@ -413,12 +413,12 @@ def start_scan(sfConfig: dict, sfModules: dict, args, loggingQueue) -> None:
if args.r:
if delim == "\t":
headers = '{0:30}{1}{2:45}{3}{4}{5}{6}'.format("Source", delim, "Type", delim, "Source Data", delim, "Data")
headers = delim.join(["Source".ljust(30), "Type".ljust(45), "Source Data", "Data"])
else:
headers = delim.join(["Source", "Type", "Source Data", "Data"])
else:
if delim == "\t":
headers = '{0:30}{1}{2:45}{3}{4}'.format("Source", delim, "Type", delim, "Data")
headers = delim.join(["Source".ljust(30), "Type".ljust(45), "Data"])
else:
headers = delim.join(["Source", "Type", "Data"])
sflib.py
+4 -8
View File
@@ -317,7 +317,7 @@ class SpiderFoot:
TypeError: arg type was invalid
"""
if not isinstance(opts, dict):
raise TypeError("opts is %s; expected dict()" % type(opts))
raise TypeError(f"opts is {type(opts)}; expected dict()")
storeopts = dict()
@@ -426,7 +426,7 @@ class SpiderFoot:
return returnOpts
if not isinstance(referencePoint['__modules__'], dict):
raise TypeError("referencePoint['__modules__'] is %s; expected dict()" % type(referencePoint['__modules__']))
raise TypeError(f"referencePoint['__modules__'] is {type(referencePoint['__modules__'])}; expected dict()")
# Module options
# A lot of mess to handle typing..
@@ -938,11 +938,7 @@ class SpiderFoot:
if not addrs:
return False
for addr in addrs:
if str(addr) == ip:
return True
return False
return any(str(addr) == ip for addr in addrs)
def safeSocket(self, host: str, port: int, timeout: int) -> 'ssl.SSLSocket':
"""Create a safe socket that's using SOCKS/TOR if it was enabled.
@@ -991,7 +987,7 @@ class SpiderFoot:
"""
if not isinstance(der_cert, bytes):
raise TypeError("der_cert is %s; expected bytes()" % type(der_cert))
raise TypeError(f"der_cert is {type(der_cert)}; expected bytes()")
return ssl.DER_cert_to_PEM_cert(der_cert)
sfscan.py
+4 -8
View File
@@ -388,11 +388,9 @@ class SpiderFootScanner():
psMod.notifyListeners(firstEvent)
# Special case.. check if an INTERNET_NAME is also a domain
if self.__targetType == 'INTERNET_NAME':
if self.__sf.isDomain(self.__targetValue, self.__config['_internettlds']):
firstEvent = SpiderFootEvent('DOMAIN_NAME', self.__targetValue,
"SpiderFoot UI", rootEvent)
psMod.notifyListeners(firstEvent)
if self.__targetType == 'INTERNET_NAME' and self.__sf.isDomain(self.__targetValue, self.__config['_internettlds']):
firstEvent = SpiderFootEvent('DOMAIN_NAME', self.__targetValue, "SpiderFoot UI", rootEvent)
psMod.notifyListeners(firstEvent)
# If in interactive mode, loop through this shared global variable
# waiting for inputs, and process them until my status is set to
@@ -569,11 +567,9 @@ class SpiderFootScanner():
if not modules_running and not queues_empty:
self.__sf.debug("Clearing queues for stalled/aborted modules.")
for mod in self.__moduleInstances.values():
try:
with suppress(Exception):
while True:
mod.incomingEventQueue.get_nowait()
except Exception:
pass
if log_status:
events_queued = ", ".join([f"{mod}: {qsize:,}" for mod, qsize in modules_waiting[:5] if qsize > 0])
sfwebui.py
+22 -26
View File
@@ -412,7 +412,7 @@ class SpiderFootWebUi:
cherrypy.response.headers['Pragma'] = "no-cache"
return self.buildExcel(rows, headings, sheetNameIndex=0)
elif filetype.lower() == 'csv':
if filetype.lower() == 'csv':
fileobj = StringIO()
parser = csv.writer(fileobj, dialect=dialect)
parser.writerow(headings)
@@ -434,8 +434,7 @@ class SpiderFootWebUi:
cherrypy.response.headers['Pragma'] = "no-cache"
return fileobj.getvalue().encode('utf-8')
else:
return self.error("Invalid export filetype.")
return self.error("Invalid export filetype.")
@cherrypy.expose
def scaneventresultexport(self: 'SpiderFootWebUi', id: str, type: str, filetype: str = "csv", dialect: str = "excel") -> str:
@@ -469,7 +468,7 @@ class SpiderFootWebUi:
return self.buildExcel(rows, ["Updated", "Type", "Module", "Source",
"F/P", "Data"], sheetNameIndex=1)
elif filetype.lower() == 'csv':
if filetype.lower() == 'csv':
fileobj = StringIO()
parser = csv.writer(fileobj, dialect=dialect)
parser.writerow(["Updated", "Type", "Module", "Source", "F/P", "Data"])
@@ -486,8 +485,7 @@ class SpiderFootWebUi:
cherrypy.response.headers['Pragma'] = "no-cache"
return fileobj.getvalue().encode('utf-8')
else:
return self.error("Invalid export filetype.")
return self.error("Invalid export filetype.")
@cherrypy.expose
def scaneventresultexportmulti(self: 'SpiderFootWebUi', ids: str, filetype: str = "csv", dialect: str = "excel") -> str:
@@ -537,7 +535,7 @@ class SpiderFootWebUi:
return self.buildExcel(rows, ["Scan Name", "Updated", "Type", "Module",
"Source", "F/P", "Data"], sheetNameIndex=2)
elif filetype.lower() == 'csv':
if filetype.lower() == 'csv':
fileobj = StringIO()
parser = csv.writer(fileobj, dialect=dialect)
parser.writerow(["Scan Name", "Updated", "Type", "Module", "Source", "F/P", "Data"])
@@ -559,8 +557,7 @@ class SpiderFootWebUi:
cherrypy.response.headers['Pragma'] = "no-cache"
return fileobj.getvalue().encode('utf-8')
else:
return self.error("Invalid export filetype.")
return self.error("Invalid export filetype.")
@cherrypy.expose
def scansearchresultexport(self: 'SpiderFootWebUi', id: str, eventType: str = None, value: str = None, filetype: str = "csv", dialect: str = "excel") -> str:
@@ -594,7 +591,7 @@ class SpiderFootWebUi:
return self.buildExcel(rows, ["Updated", "Type", "Module", "Source",
"F/P", "Data"], sheetNameIndex=1)
elif filetype.lower() == 'csv':
if filetype.lower() == 'csv':
fileobj = StringIO()
parser = csv.writer(fileobj, dialect=dialect)
parser.writerow(["Updated", "Type", "Module", "Source", "F/P", "Data"])
@@ -608,8 +605,7 @@ class SpiderFootWebUi:
cherrypy.response.headers['Pragma'] = "no-cache"
return fileobj.getvalue().encode('utf-8')
else:
return self.error("Invalid export filetype.")
return self.error("Invalid export filetype.")
@cherrypy.expose
def scanexportjsonmulti(self: 'SpiderFootWebUi', ids: str) -> str:
@@ -1106,28 +1102,28 @@ class SpiderFootWebUi:
if str(token) != str(self.token):
return self.error(f"Invalid token ({token})")
if configFile: # configFile seems to get set even if a file isn't uploaded
if configFile.file:
# configFile seems to get set even if a file isn't uploaded
if configFile and configFile.file:
try:
contents = configFile.file.read()
if isinstance(contents, bytes):
contents = contents.decode('utf-8')
try:
tmp = dict()
for line in contents.split("\n"):
if "=" not in line:
continue
tmp = dict()
for line in contents.split("\n"):
if "=" not in line:
continue
opt_array = line.strip().split("=")
if len(opt_array) == 1:
opt_array[1] = ""
opt_array = line.strip().split("=")
if len(opt_array) == 1:
opt_array[1] = ""
tmp[opt_array[0]] = '='.join(opt_array[1:])
tmp[opt_array[0]] = '='.join(opt_array[1:])
allopts = json.dumps(tmp).encode('utf-8')
except Exception as e:
return self.error(f"Failed to parse input file. Was it generated from SpiderFoot? ({e})")
allopts = json.dumps(tmp).encode('utf-8')
except Exception as e:
return self.error(f"Failed to parse input file. Was it generated from SpiderFoot? ({e})")
# Reset config to default
if allopts == "RESET":
spiderfoot/correlation.py
+12 -16
View File
@@ -422,10 +422,7 @@ class SpiderFootCorrelator:
if "." in field:
key, field = field.split(".")
for subevent in event[key]:
if self.event_keep(subevent, field, patterns, patterntype):
return True
return False
return any(self.event_keep(subevent, field, patterns, patterntype) for subevent in event[key])
value = event[field]
@@ -638,10 +635,9 @@ class SpiderFootCorrelator:
except Exception:
pass
if rule['match_method'] == 'exact':
if event_data in reference:
self.log.debug(f"found exact match: {event_data} in {reference}")
return True
if rule['match_method'] == 'exact' and event_data in reference:
self.log.debug(f"found exact match: {event_data} in {reference}")
return True
if rule['match_method'] == 'contains':
for r in reference:
@@ -675,7 +671,7 @@ class SpiderFootCorrelator:
# delete the bucket if there are no events > collection 0
if pluszerocount == 0:
del(buckets[bucket])
del (buckets[bucket])
def analysis_first_collection_only(self, rule: dict, buckets: dict) -> None:
"""analysis_first_collection_only TBD
@@ -699,13 +695,13 @@ class SpiderFootCorrelator:
delete = True
break
if delete:
del(buckets[bucket])
del (buckets[bucket])
# Remove buckets with collection > 0 values
for bucket in list(buckets.keys()):
for e in buckets[bucket]:
if e['_collection'] > 0:
del(buckets[bucket])
del (buckets[bucket])
break
def analysis_outlier(self, rule: dict, buckets: dict) -> None:
@@ -722,7 +718,7 @@ class SpiderFootCorrelator:
if len(list(countmap.keys())) == 0:
for bucket in list(buckets.keys()):
del(buckets[bucket])
del (buckets[bucket])
return
total = float(sum(countmap.values()))
@@ -733,7 +729,7 @@ class SpiderFootCorrelator:
if avgpct < rule.get('noisy_percent', 10):
self.log.debug(f"Not correlating because the average percent is {avgpct} (too anomalous)")
for bucket in list(buckets.keys()):
del(buckets[bucket])
del (buckets[bucket])
return
# Figure out which buckets don't contain outliers and delete them
@@ -743,7 +739,7 @@ class SpiderFootCorrelator:
delbuckets.append(bucket)
for bucket in set(delbuckets):
del(buckets[bucket])
del (buckets[bucket])
def analysis_threshold(self, rule: dict, buckets: dict) -> None:
"""analysis_treshold TBD
@@ -769,14 +765,14 @@ class SpiderFootCorrelator:
# Delete the bucket of events if it didn't meet the
# analysis criteria.
if bucket in buckets:
del(buckets[bucket])
del (buckets[bucket])
continue
# If we're only looking at the number of times the requested
# field appears in the bucket...
uniques = len(list(countmap.keys()))
if uniques < rule.get('minimum', 0) or uniques > rule.get('maximum', 999999999):
del(buckets[bucket])
del (buckets[bucket])
def analyze_field_scope(self, field: str) -> list:
"""Analysis field scope.
spiderfoot/db.py
+6 -9
View File
@@ -1382,9 +1382,8 @@ class SpiderFootDb:
if not isinstance(sfEvent.module, str):
raise TypeError(f"sfEvent.module is {type(sfEvent.module)}; expected str()") from None
if not sfEvent.module:
if sfEvent.eventType != "ROOT":
raise ValueError("sfEvent.module is empty") from None
if not sfEvent.module and sfEvent.eventType != "ROOT":
raise ValueError("sfEvent.module is empty") from None
if not isinstance(sfEvent.confidence, int):
raise TypeError(f"sfEvent.confidence is {type(sfEvent.confidence)}; expected int()") from None
@@ -1404,9 +1403,8 @@ class SpiderFootDb:
if not 0 <= sfEvent.risk <= 100:
raise ValueError(f"sfEvent.risk value is {type(sfEvent.risk)}; expected 0 - 100") from None
if not isinstance(sfEvent.sourceEvent, SpiderFootEvent):
if sfEvent.eventType != "ROOT":
raise TypeError(f"sfEvent.sourceEvent is {type(sfEvent.sourceEvent)}; expected str()") from None
if not isinstance(sfEvent.sourceEvent, SpiderFootEvent) and sfEvent.eventType != "ROOT":
raise TypeError(f"sfEvent.sourceEvent is {type(sfEvent.sourceEvent)}; expected str()") from None
if not isinstance(sfEvent.sourceEventHash, str):
raise TypeError(f"sfEvent.sourceEventHash is {type(sfEvent.sourceEventHash)}; expected str()") from None
@@ -1417,9 +1415,8 @@ class SpiderFootDb:
storeData = sfEvent.data
# truncate if required
if isinstance(truncateSize, int):
if truncateSize > 0:
storeData = storeData[0:truncateSize]
if isinstance(truncateSize, int) and truncateSize > 0:
storeData = storeData[0:truncateSize]
# retrieve scan results
qry = "INSERT INTO tbl_scan_results \
spiderfoot/event.py
+4 -6
View File
@@ -226,9 +226,8 @@ class SpiderFootEvent():
if not isinstance(module, str):
raise TypeError(f"module is {type(module )}; expected str()")
if not module:
if self.eventType != "ROOT":
raise ValueError("module is empty")
if not module and self.eventType != "ROOT":
raise ValueError("module is empty")
self._module = module
@@ -296,9 +295,8 @@ class SpiderFootEvent():
'source': ''
}
if self.sourceEvent is not None:
if self.sourceEvent.data is not None:
evtDict['source'] = self.sourceEvent.data
if self.sourceEvent is not None and self.sourceEvent.data is not None:
evtDict['source'] = self.sourceEvent.data
return evtDict
spiderfoot/plugin.py
+7 -10
View File
@@ -329,12 +329,10 @@ class SpiderFootPlugin():
eventName = sfEvent.eventType
eventData = sfEvent.data
if self.__outputFilter__:
# Be strict about what events to pass on, unless they are
# the ROOT event or the event type of the target.
if eventName not in ('ROOT', self.getTarget().targetType):
if eventName not in self.__outputFilter__:
return
# Be strict about what events to pass on, unless they are
# the ROOT event or the event type of the target.
if self.__outputFilter__ and eventName not in ['ROOT', self.getTarget().targetType, self.__outputFilter__]:
return
storeOnly = False # Under some conditions, only store and don't notify
@@ -360,10 +358,9 @@ class SpiderFootPlugin():
prevEvent = sfEvent.sourceEvent
while prevEvent is not None:
if prevEvent.sourceEvent is not None:
if prevEvent.sourceEvent.eventType == sfEvent.eventType and prevEvent.sourceEvent.data.lower() == eventData.lower():
storeOnly = True
break
if prevEvent.sourceEvent is not None and prevEvent.sourceEvent.eventType == sfEvent.eventType and prevEvent.sourceEvent.data.lower() == eventData.lower():
storeOnly = True
break
prevEvent = prevEvent.sourceEvent
# output to queue if applicable
test/unit/spiderfoot/test_spiderfoothelpers.py
+1 -1
View File
@@ -178,7 +178,7 @@ class TestSpiderFootHelpers(unittest.TestCase):
)
def test_extractLinksFromHtml_invalid_url_should_raise_TypeError(self):
invalid_types = [None, bytes(), list(), dict()]
invalid_types = [None, bytes(), list(), dict()]
for invalid_type in invalid_types:
with self.subTest(invalid_type=invalid_type):
with self.assertRaises(TypeError):