CSEC_PUBLIC/securityonion
1
0
Fork 0
mirror of https://github.com/Security-Onion-Solutions/securityonion.git synced 2025-12-06 09:12:45 +01:00
Code Issues Packages Projects Releases Wiki Activity

Flake8 linting + isInJson tail recursion update

Browse Source
This commit is contained in:
Elijah Gibson
2023-12-18 15:58:16 -05:00
committed by GitHub
parent 7d6f8d922b
commit fb5ee6b9e9
1 changed files with 19 additions and 12 deletions
Download Patch File Download Diff File Expand all files Collapse all files

31
salt/sensoroni/files/analyzers/malwarebazaar/malwarebazaar.py
View File

@@ -25,18 +25,18 @@ def sendReq(meta, query):
return response.json()
def isInJson(data, target_string, maxdepth):
def isInJson(data, target_string, maxdepth=1000, tail=0):
# searches a JSON object for an occurance of a string
# recursively.
# depth limiter (arbitrary value of 1000)
if maxdepth > 1000:
# depth limiter (arbitrary default value of 1000)
if tail > maxdepth:
return False
if isinstance(data, dict):
for key, value in data.items():
if isinstance(value, (dict, list)):
# recursive call
if isInJson(value, target_string, maxdepth + 1):
if isInJson(value, target_string, maxdepth, tail + 1):
return True
elif isinstance(value, str) and target_string in value.lower():
# found target string
@@ -46,7 +46,7 @@ def isInJson(data, target_string, maxdepth):
for item in data:
if isinstance(item, (dict, list)):
# recursive call
if isInJson(item, target_string, maxdepth + 1):
if isInJson(item, target_string, maxdepth, tail + 1):
return True
elif isinstance(item, str) and target_string in item.lower():
# found target string
@@ -56,7 +56,8 @@ def isInJson(data, target_string, maxdepth):
def prepareResults(raw):
# parse raw API response, gauge threat level and return status and a short summary
# parse raw API response, gauge threat level
# and return status and a short summary
if raw == {}:
status = 'caution'
summary = 'internal_failure'
@@ -72,7 +73,8 @@ def prepareResults(raw):
elif 'YOROI_YOMI' in vendor_data:
summary = vendor_data['YOROI_YOMI']['detection']
# gauge vendors to determine an approximation of status, normalized to a value out of 100
# gauge vendors to determine an approximation of status,
# normalized to a value out of 100
# only updates score if it finds a higher indicator value
score = 0
vendor_info_list = [
@@ -81,8 +83,10 @@ def prepareResults(raw):
('DocGuard', 'alertlevel', lambda x: int(x) * 10),
('YOROI_YOMI', 'score', lambda x: int(float(x)) * 100),
('Inquest', 'verdict', lambda x: 100 if x == 'MALICIOUS' else 0),
('ReversingLabs', 'status', lambda x: 100 if x == 'MALICIOUS' else 0),
('Spamhaus_HBL', 'detection', lambda x: 100 if x == 'MALICIOUS' else 0),
('ReversingLabs', 'status',
lambda x: 100 if x == 'MALICIOUS' else 0),
('Spamhaus_HBL', 'detection',
lambda x: 100 if x == 'MALICIOUS' else 0),
]
for vendor, key, transform in vendor_info_list:
if vendor in vendor_data and key in vendor_data[vendor]:
@@ -116,14 +120,17 @@ def analyze(input):
meta = helpers.loadMetadata(__file__)
helpers.checkSupportedType(meta, data["artifactType"])
if (data['artifactType'] == 'tlsh' or data['artifactType'] == 'gimphash' or data['artifactType'] == 'telfhash'):
# To get accurate reporting for TLSH, telfhash and gimphash, we deem it necessary to query
if (data['artifactType'] == 'tlsh' or data['artifactType'] == 'gimphash'
or data['artifactType'] == 'telfhash'):
# To get accurate reporting for TLSH, telfhash and gimphash,
# we deem it necessary to query
# twice for the sake of retrieving more specific data.
initialQuery = buildReq(data['artifactType'], data['value'])
initialRaw = sendReq(meta, initialQuery)
# To prevent double-querying when a tlsh/gimphash is invalid, this if statement is necessary.
# To prevent double-querying when a tlsh/gimphash is invalid,
# this if statement is necessary.
if initialRaw['query_status'] == 'ok':
# Setting artifactType and value to our new re-query arguments
# to get a more detailed report.